package com.xiguanlezz.cn.common.remote.client;

import com.xiguanlezz.cn.api.common.Constants;
import com.xiguanlezz.cn.api.remote.RequestCallBack;
import com.xiguanlezz.cn.api.remote.RequestFuture;
import com.xiguanlezz.cn.api.remote.request.HealthCheckRequest;
import com.xiguanlezz.cn.api.remote.request.Request;
import com.xiguanlezz.cn.api.remote.response.HealthCheckResponse;
import com.xiguanlezz.cn.api.remote.response.Response;
import com.xiguanlezz.cn.common.lifecycle.Closeable;
import com.xiguanlezz.cn.common.remote.ConnectionType;
import com.xiguanlezz.cn.common.remote.PayloadRegistry;
import com.xiguanlezz.cn.common.utils.InternetAddressUtil;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.math.NumberUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicReference;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * @Author：习惯了种种
 * @CreateDate：2024/11/19 16:16
 * @Description：RPC客户端通信对象
 */
@Slf4j
public abstract class RpcClient implements Closeable {
    private RpcClientConfig rpcClientConfig; /* RPC客户端配置类 */
    private ServerListFactory serverListFactory; /* Nacos服务端地址列表管理类 */
    @Getter
    // 并发安全的客户端状态
    protected AtomicReference<RpcClientStatus> rpcClientStatus;
    // 封装了客户端连接，可用于发送请求
    protected volatile Connection currentConnection;
    // 客户端事件执行器，用于连接重建和服务端健康检查
    private ScheduledExecutorService clientEventExecutor;
    // 存放重连上下文的阻塞队列，当队列中有数据时，说明此时可能网络存在波动，连接需要重建
    private BlockingQueue<ReconnectContext> reconnectionSignal = new ArrayBlockingQueue<>(1);
    // 上次服务端的活跃时间
    private long lastActiveTimeStamp;

    // 当客户端访问服务端的时候，如果构建连接成功了，就会创建一个连接事件，存放到这个阻塞队列中
    // 如果连接断开了，也会创建一个断开连接事件，存放到这个阻塞队列中
    protected BlockingQueue<ConnectionEvent> eventLinkedBlockingQueue = new LinkedBlockingQueue<>(); /* 存放的是连接事件的阻塞队列 */

    // 客户端连接事件监听器
    protected List<ConnectionEventListener> connectionEventListeners = new ArrayList<>();

    // 用于处理服务端主动发来的请求
    protected List<ServerRequestHandler> serverRequestHandlers = new ArrayList<>();


    private static final Pattern EXCLUDE_PROTOCOL_PATTERN = Pattern.compile("(?<=\\w{1,5}://)(.*)");

    static {
        PayloadRegistry.init();
    }


    public RpcClient(RpcClientConfig rpcClientConfig) {
        this.rpcClientConfig = rpcClientConfig;
        rpcClientStatus = new AtomicReference<>(RpcClientStatus.WAIT_INIT);
    }

    public RpcClient(RpcClientConfig rpcClientConfig, ServerListFactory serverListFactory) {
        this(rpcClientConfig);
        this.serverListFactory = serverListFactory;
    }

    public Response request(Request request) throws Exception {
        return request(request, rpcClientConfig.timeOutMills());
    }

    private Response request(Request request, long timeOutMills) throws Exception {
        int retryTimes = 0; /* 重试次数 */
        Response response; /* 接收的响应对象 */
        Throwable exceptionThrow = null; /* 接收的异常信息 */
        long start = System.currentTimeMillis(); /* 操作的开始时间 */
        while (retryTimes < rpcClientConfig.retryTimes() && (timeOutMills <= 0 || System.currentTimeMillis() < start + timeOutMills)) {
            try {
                if (Objects.isNull(currentConnection) || !isRunning()) {
                    throw new RuntimeException("Client not connected, current status:" + rpcClientStatus.get());
                }
                response = currentConnection.request(request, timeOutMills);
                if (Objects.isNull(response)) {
                    throw new RuntimeException("Unknown Exception.");
                }

                // 更新上次服务端活跃的时间戳
                lastActiveTimeStamp = System.currentTimeMillis();
                return response;
            } catch (Exception e) {
                exceptionThrow = e;
                log.error("Send request fail, request = {}, retryTimes = {}, errorMessage = {}",
                        request,
                        retryTimes,
                        e.getMessage());
            }
            retryTimes++;
        }
        // 走到这里意味着连接出问题了，这时候就要把客户端状态更新为不健康状态，等待客户端异步连接服务器成功
        if (rpcClientStatus.compareAndSet(RpcClientStatus.RUNNING, RpcClientStatus.UNHEALTHY)) {
            switchServerAsyncOnRequestFail();
        }
        if (Objects.nonNull(exceptionThrow)) {
            throw new RuntimeException(exceptionThrow.getMessage(), exceptionThrow);
        } else {
            throw new RuntimeException("Request fail, unknown Error");
        }
    }

    public RequestFuture requestFuture(Request request) throws Exception {
        if (Objects.isNull(currentConnection) || !isRunning()) {
            throw new RuntimeException("Client not connected, current status:" + rpcClientStatus.get());
        }
        return currentConnection.requestFuture(request);
    }

    public void asyncRequest(Request request, RequestCallBack requestCallBack) throws Exception {
        if (Objects.isNull(currentConnection) || !isRunning()) {
            throw new RuntimeException("Client not connected, current status:" + rpcClientStatus.get());
        }
        currentConnection.asyncRequest(request, requestCallBack);
    }

    /**
     * 注册事件监听器方法
     */
    public synchronized void registerConnectionListener(ConnectionEventListener connectionEventListener) {
        log.info("[{}] registry connection listener to current client:{}",
                rpcClientConfig.name(),
                connectionEventListener.getClass().getName());
        this.connectionEventListeners.add(connectionEventListener);
    }

    /**
     * 处理服务端请求
     */
    protected Response handleServerRequest(final Request request) {
        log.error("[{}] Receive server push request, request = {}, requestId = {}",
                rpcClientConfig.name(), request.getClass().getSimpleName(), request.getRequestId());
        lastActiveTimeStamp = System.currentTimeMillis();
        for (ServerRequestHandler serverRequestHandler : serverRequestHandlers) {
            try {
                Response response = serverRequestHandler.requestReply(request, currentConnection);
                if (Objects.nonNull(response)) {
                    log.error("[{}] Ack server push request, request = {}, requestId = {}",
                            rpcClientConfig.name(), request.getClass().getSimpleName(), request.getRequestId());
                    return response;
                }
            } catch (Exception e) {
                log.error("[{}] HandleServerRequest:{}, errorMessage = {}",
                        rpcClientConfig.name(), serverRequestHandler.getClass().getName(), e.getMessage());
                throw e;
            }
        }
        return null;
    }

    /**
     * 注册服务端请求处理器
     */
    public synchronized void registerServerRequestHandler(ServerRequestHandler serverRequestHandler) {
        log.error("[{}] register server push request handler:{}", rpcClientConfig.name(),
                serverRequestHandler.getClass().getName());
        this.serverRequestHandlers.add(serverRequestHandler);
    }

    /**
     * 客户端健康检查方法
     */
    private boolean healthCheck() {
        HealthCheckRequest healthCheckRequest = new HealthCheckRequest();
        if (Objects.isNull(currentConnection)) {
            return false;
        }
        try {
            ThreadLocalRandom threadLocalRandom = ThreadLocalRandom.current();
            int healthCheckRetryTimes = rpcClientConfig.healthCheckRetryTimes();
            while (healthCheckRetryTimes > 0) {
                Response response = currentConnection.request(healthCheckRequest, rpcClientConfig.healthCheckTimeOut());
                if (response instanceof HealthCheckResponse) {
                    log.info("Health check successfully!");
                }
                --healthCheckRetryTimes;
                TimeUnit.MILLISECONDS.sleep(threadLocalRandom.nextInt(500));
                return Objects.nonNull(response) && response.isSuccess();
            }
        } catch (Exception e) {
        }
        return false;
    }

    /**
     * 外部调用RpcClient时，会先执行serverListFactory方法，此时客户端状态会从等待初始化状态变为初始化状态
     */
    public RpcClient serverListFactory(ServerListFactory serverListFactory) {
        if (!isWaitInitiated()) {
            return this;
        }
        this.serverListFactory = serverListFactory;
        rpcClientStatus.compareAndSet(RpcClientStatus.WAIT_INIT, RpcClientStatus.INITIALIZED);
        log.info("[{}] RpcClient init, ServerListFactory = {}", rpcClientConfig.name(), serverListFactory.getClass().getName());
        return this;
    }

    public void start() {
        // 将客户端状态从初始化状态改为启动中状态
        boolean success = rpcClientStatus.compareAndSet(RpcClientStatus.INITIALIZED, RpcClientStatus.STARTING);
        if (!success) {
            return;
        }

        // 初始化客户端事件执行器
        // 第一个线程用于心跳检测、断线重连
        // 第二个线程用于自动执行断线重连后的重试操作
        clientEventExecutor = new ScheduledThreadPoolExecutor(2, task -> {
            Thread t = new Thread(task);
            t.setName("com.alibaba.nacos.client.remote.worker");
            t.setDaemon(true);
            return t;
        });
        clientEventExecutor.submit(() -> {
            // 1. 如果阻塞队列中不存在重连上下文，说明此时客户端和服务端之间的连接是正常的，只需要发送健康检查请求，探测服务端是否存活
            //    当服务端CPU使用过大时，虽然连接是存活的，但是可能没空响应健康检查，也算是服务端挂掉了
            // 2. 如果阻塞队列中存在重连上下文，根据重连上下文对象向服务端发起建立连接即可
            while (true) {
                try {
                    ReconnectContext reconnectContext = reconnectionSignal.poll(rpcClientConfig.connectionKeepAlive(), TimeUnit.MILLISECONDS);
                    if (Objects.isNull(reconnectContext)) {
                        if (System.currentTimeMillis() - lastActiveTimeStamp < rpcClientConfig.connectionKeepAlive()) {
                            // 如果没到连接的探活间隔直接走到下一次循环
                            continue;
                        }

                        if (healthCheck()) {
                            // 如果健康检查成果，重置下上次服务端的活跃时间，走下一个循环集合
                            lastActiveTimeStamp = System.currentTimeMillis();
                            continue;
                        }
                        if (Objects.isNull(currentConnection)) {
                            // 当程序初次启动时，客户端和服务端还没建立连接时，肯定会走到这里，因为客户端启动方法和客户端事件执行器之间是并行执行的
                            continue;
                        }
                        RpcClientStatus rpcClientStatus = RpcClient.this.rpcClientStatus.get();
                        if (RpcClientStatus.SHUTDOWN.equals(rpcClientStatus)) {
                            // 如果客户端已关闭直接退出循环即可
                            break;
                        }
                        boolean updateStatusSuccess = RpcClient.this.rpcClientStatus.compareAndSet(rpcClientStatus, RpcClientStatus.UNHEALTHY);
                        if (updateStatusSuccess) {
                            reconnectContext = new ReconnectContext(null, false);
                        } else {
                            continue;
                        }
                    }

                    // 重新建立连接
                    reconnect(reconnectContext.serverInfo, reconnectContext.onRequestFail);
                } catch (Exception e) {
                    // do nothing...
                }
            }
        });

        clientEventExecutor.submit(() -> {
            while (!clientEventExecutor.isTerminated() && !clientEventExecutor.isShutdown()) {
                ConnectionEvent event = null;
                try {
                    // 阻塞获取连接事件
                    event = eventLinkedBlockingQueue.take();
                    if (event.isConnected()) {
                        // 连接事件通知
                        notifyConnected(event.connection);
                    } else {
                        // 断连事件通知
                        notifyDisConnected(event.connection);
                    }
                } catch (Exception e) {
                }
            }
        });

        Connection connectToServer = null;
        ServerInfo serverInfo = currentRpcServer();
        int startUpRetryTimes = rpcClientConfig.retryTimes();
        try {
            while (startUpRetryTimes > 0 && Objects.isNull(connectToServer)) {
                try {
                    log.info("[{}] try to connect to server on start up, server: {}", rpcClientConfig.name(), serverInfo);
                    connectToServer = connectToServer(serverInfo);
                    --startUpRetryTimes;
                    // 选取下个地址重新开始建立连接
                    serverInfo = nextRpcServer();
                } catch (Exception e) {
                    log.error("[{}] fail to connect to server on start up, error message = {}, start up retry times left: {}",
                            rpcClientConfig.name(), e.getMessage(), startUpRetryTimes, e);
                }
            }
            if (Objects.nonNull(connectToServer)) {
                // 将客户端状态置为运行状态
                rpcClientStatus.set(RpcClientStatus.RUNNING);
                currentConnection = connectToServer;
            } else {
                // 连接失败，需要和定时任务线程通信
                switchServerAsync();
            }
        } catch (Exception e) {
            log.error("[{}] fail to connect to server on start up...", rpcClientConfig.name());
        }
    }

    @Override
    public void shutdown() throws Exception {
        log.info("Shutdown rpc client, set status to shutdown");
        rpcClientStatus.set(RpcClientStatus.SHUTDOWN);
        log.info("Shutdown client event executor " + clientEventExecutor);
        if (Objects.nonNull(clientEventExecutor)) {
            clientEventExecutor.shutdownNow();
        }
        closeConnection(currentConnection);
    }

    protected void notifyConnected(Connection connection) {
        // 首先判断客户端连接事件监听器中是否有数据，如果没有就直接退出当前方法
        if (connectionEventListeners.isEmpty()) {
            return;
        }
        log.info("[{}] notify connected event to listeners.", rpcClientConfig.name());
        // 遍历并执行每一个监听器的onConnected方法
        for (ConnectionEventListener connectionEventListener : connectionEventListeners) {
            try {
                connectionEventListener.onConnected(connection);
            } catch (Throwable throwable) {
                log.error("[{}] notify connect listener error, listener = {}",
                        rpcClientConfig.name(),
                        connectionEventListener.getClass().getName());
            }
        }
    }

    protected void notifyDisConnected(Connection connection) {
        // 首先判断客户端连接事件监听器中是否有数据，如果没有就直接退出当前方法
        if (connectionEventListeners.isEmpty()) {
            return;
        }
        log.info("[{}] notify disconnected event to listeners.", rpcClientConfig.name());
        // 遍历并执行每一个监听器的onDisConnect方法
        for (ConnectionEventListener connectionEventListener : connectionEventListeners) {
            try {
                connectionEventListener.onDisConnect(connection);
            } catch (Throwable throwable) {
                log.error("[{}] notify disconnect listener error, listener = {}",
                        rpcClientConfig.name(),
                        connectionEventListener.getClass().getName());
            }
        }
    }

    public void switchServerAsync() {
        switchServerAsync(null, false);
    }

    public void switchServerAsyncOnRequestFail() {
        switchServerAsync(null, true);
    }

    private void switchServerAsync(ServerInfo recommendServerInfo, boolean onRequestFail) {
        reconnectionSignal.offer(new ReconnectContext(recommendServerInfo, onRequestFail));
    }

    private void reconnect(ServerInfo recommendServerInfo, boolean onRequestFail) {
        try {
            // 使用原子引用感觉没必要，本身定时任务单线程执行，而且阻塞队列也保证了重连上下文的线程安全性
            AtomicReference<ServerInfo> recommendServer = new AtomicReference<>(recommendServerInfo);
            // 当onRequestFail为true时，先进行一次探活
            if (onRequestFail && healthCheck()) {
                // 若服务端存活则更新客户端状态
                rpcClientStatus.set(RpcClientStatus.RUNNING);
                return;
            }

            boolean switchSuccess = false; /* 重连成功的标记 */
            int reConnectTimes = 0; /* 重连次数 */
            int retryTurns = 0; /* 重试轮数，如果重连次数为6，服务端地址列表中有2个可用地址，最大轮数即为6 / 2 = 3 */
            Exception lastException = null;
            // 只有客户端关闭或者连接成功才会退出循环
            while (!switchSuccess && !isShutdown()) {
                ServerInfo serverInfo = null;
                try {
                    serverInfo = Objects.nonNull(recommendServer.get()) ? recommendServer.get() : nextRpcServer();
                    // 连接Nacos服务端
                    Connection newConnection = connectToServer(serverInfo);
                    if (Objects.nonNull(newConnection)) {
                        if (Objects.nonNull(currentConnection)) {
                            // 抛弃并关闭旧连接
                            currentConnection.setAbandon(true);
                            closeConnection(currentConnection);
                        }
                        currentConnection = newConnection;
                        rpcClientStatus.set(RpcClientStatus.RUNNING);
                        switchSuccess = true;
                        // 往阻塞队列中加入连接事件
                        eventLinkedBlockingQueue.add(new ConnectionEvent(ConnectionEvent.CONNECTED, currentConnection));
                        return;
                    }
                    // 说明重连失败。此时如果客户端是关闭状态，需要关闭底层连接
                    if (isShutdown()) {
                        closeConnection(currentConnection);
                    }
                    lastException = null;
                } catch (Exception e) {
                    lastException = e;
                } finally {
                    recommendServer.set(null);
                }

                // 走到这里说明本次连接没有成功，接下来要更换一个新的服务地址再次重试了
                // 所以先判断一下服务地址管理器是否为空
                if (CollectionUtils.isEmpty(RpcClient.this.serverListFactory.getServerList())) {
                    throw new Exception("server list is empty");
                }


                // reConnectTimes % RpcClient.this.serverListFactory.getServerList().size() == 0就是重试的次数和服务地址管理器中的服务地址做除法，并且被整除了
                // 更新重试轮数，应该是基于减少日志量的考虑
                if (reConnectTimes > 0 && reConnectTimes % RpcClient.this.serverListFactory.getServerList().size() == 0) {
                    log.error("[{}] fail to connect server, after trying {} times, last try server is {}, error = {}",
                            rpcClientConfig.name(),
                            reConnectTimes,
                            serverInfo,
                            lastException == null ? "unknown" : lastException);
                    if (Integer.MAX_VALUE == retryTurns) {
                        // 重试轮数达到上限就把其重置为50
                        retryTurns = 50;
                    } else {
                        retryTurns++;
                    }
                }

                // 自增重试次数
                reConnectTimes++;
                try {
                    // 判断客户端的状态，如果不是运行状态，说明客户端仍然没有成功连接到服务端
                    if (!isRunning()) {
                        // 重试轮数也被用来实现退避重试
                        Thread.sleep(Math.min(retryTurns + 1, 50) * 100L);
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }

                // 如果客户端是关闭状态，则记录日志，下次会直接循环，不再重试
                if (isShutdown()) {
                    log.error("[{}] client is shutdown, stop reconnect to server", rpcClientConfig.name());
                }
            }
        } catch (Exception e) {
            log.error("[{}] fail to reconnect to server, error is {}", rpcClientConfig.name(), e);
        }
    }

    private void closeConnection(Connection connection) {
        if (Objects.nonNull(connection)) {
            connection.close();
            // 往阻塞队列中加入断连事件
            eventLinkedBlockingQueue.add(new ConnectionEvent(ConnectionEvent.DISCONNECTED, connection));
        }
    }

    /**
     * 判断客户端是否是等待初始化状态
     */
    public boolean isWaitInitiated() {
        return this.rpcClientStatus.get() == RpcClientStatus.WAIT_INIT;
    }

    /**
     * 判断客户端是否是运行状态
     */
    public boolean isRunning() {
        return this.rpcClientStatus.get() == RpcClientStatus.RUNNING;
    }

    /**
     * 判断客户端是否是关闭状态
     */
    public boolean isShutdown() {
        return this.rpcClientStatus.get() == RpcClientStatus.SHUTDOWN;
    }

    public Map<String, String> getLabels() {
        return rpcClientConfig.labels();
    }

    protected ServerInfo nextRpcServer() {
        String serverAddress = serverListFactory.genNextServer();
        return resolveServerInfo(serverAddress);
    }

    protected ServerInfo currentRpcServer() {
        String serverAddress = serverListFactory.getCurrentServer();
        return resolveServerInfo(serverAddress);
    }

    private ServerInfo resolveServerInfo(String serverAddress) {
        // 使用正则表达式移除服务地址中的协议部分
        Matcher matcher = EXCLUDE_PROTOCOL_PATTERN.matcher(serverAddress);
        if (matcher.find()) {
            serverAddress = matcher.group(1);
        }
        // 从服务地址中解析出IP地址和端口，服务地址既可能是IPV4也可能是IPV6
        String[] ipPortTuple = InternetAddressUtil.splitIPPortStr(serverAddress);
        String ip = ipPortTuple[0];
        int port = 0;
        int defaultPort = Integer.parseInt(System.getProperty("nacos.server.port", "8848"));
        if (ArrayUtils.isNotEmpty(ipPortTuple) && ipPortTuple.length == 2) {
            port = NumberUtils.toInt(ipPortTuple[1], defaultPort);
        }
        // 把ip和port封装在ServerInfo对象中
        return new ServerInfo(ip, port);
    }

    /**
     * RPC通信类型
     */
    public abstract ConnectionType getConnectionType();

    /**
     * 服务注册端口相较于Nacos服务端主端口的偏移量
     */
    public abstract int rpcPortOffset();

    public abstract Connection connectToServer(ServerInfo serverInfo) throws Exception;

    /**
     * 连接事件实体类
     */
    public static class ConnectionEvent {
        public static final int CONNECTED = 1;
        public static final int DISCONNECTED = 0;

        int eventType;
        Connection connection;

        public ConnectionEvent(int eventType, Connection connection) {
            this.eventType = eventType;
            this.connection = connection;
        }

        public boolean isConnected() {
            return eventType == CONNECTED;
        }

        public boolean isDisConnected() {
            return eventType == DISCONNECTED;
        }
    }

    /**
     * 连接重建上下文对象
     */
    static class ReconnectContext {
        public ReconnectContext(ServerInfo serverInfo, boolean onRequestFail) {
            this.onRequestFail = onRequestFail;
            this.serverInfo = serverInfo;
        }

        boolean onRequestFail;
        ServerInfo serverInfo;
    }


    @Setter
    @Getter
    public static class ServerInfo {
        protected String serverIp;
        protected int serverPort;

        public ServerInfo() {
        }

        public ServerInfo(String serverIp, int serverPort) {
            this.serverPort = serverPort;
            this.serverIp = serverIp;
        }

        /**
         * get address, ip:port.
         *
         * @return address.
         */
        public String getAddress() {
            return serverIp + Constants.COLON + serverPort;
        }


        @Override
        public String toString() {
            return "{serverIp = '" + serverIp + '\'' + ", server main port = " + serverPort + '}';
        }
    }

}
